# -*- encoding: utf-8 -*-

"""
------------------------------------------
@File       : deal_pic_with_queue_pro.py
@Author     : maixiaochai
@Email      : maixiaochai@outlook.com
@CreatedOn  : 2021/6/16 14:07
------------------------------------------
[2021-06-15]
以下高级的写法，可以解决以下三个问题
    1）线程饥饿造成的CPU时间浪费问题
    2）无法得知何时停止的问题,或者低效的方法导致CPU时间的浪费
    3）前后 worker效率不一致导致相连队列不断增，耗尽内存的问题

原作者代码：https://github.com/bslatkin/effectivepython/blob/master/example_code/item_55.py
"""
from queue import Queue
from threading import Thread, Lock
from time import sleep


class CloseableQueue(Queue):
    """
        比较高级的队列用法
        Tips:
            Queue.get()会持续阻塞，直到队列中 put数据才返回
    """
    SENTINEL = object()

    def close(self):
        self.put(self.SENTINEL)

    def __iter__(self):
        while True:
            # 因为queue.get会持续阻塞，所以，在流程中并不会造成CPU时间的浪费
            item = self.get()
            try:
                if item is self.SENTINEL:
                    return  # 让线程退出

                yield item

            # 在python的异常捕捉中，不论前边出错与否，finally一定会执行，这里是确保 self.task_done()一定会被执行
            finally:
                self.task_done()


class StoppableWorker(Thread):
    """
        比较高级的 Worker
    """

    def __init__(self, func, in_queue, out_queue):
        super().__init__()
        self.func = func
        self.in_queue = in_queue
        self.out_queue = out_queue
        self.lock = Lock()

    def run(self):
        # todo: 为什么这里不用锁 —— with lock: ...
        for item in self.in_queue:
            with self.lock:
                result = self.func(item)
                self.out_queue.put(result)


def start_threads(count, *args):
    threads = [StoppableWorker(*args) for _ in range(count)]
    for thread in threads:
        thread.start()

    return threads


def stop_thread(closable_queue, threads):
    for _ in threads:
        closable_queue.close()

    # 实际上意味着等到队列为空，再执行别的操作
    closable_queue.join()

    # 等待所有线程结束再推出
    for thread in threads:
        thread.join()


def download(d):
    sleep(1)
    result = f"{d}-d"
    print(result)

    return result


def resize(r):
    sleep(0.5)
    result = f"{r}-r"
    print(result)

    return result


def upload(u):
    sleep(0.5)
    result = f"{u}-u"
    print(result)

    return result


def main():
    download_queue = CloseableQueue()
    resize_queue = CloseableQueue()
    upload_queue = CloseableQueue()
    done_queue = CloseableQueue()

    download_threads = start_threads(3, download, download_queue, resize_queue)
    resize_threads = start_threads(4, resize, resize_queue, upload_queue)
    upload_threads = start_threads(5, upload, upload_queue, done_queue)

    for i in range(100):
        download_queue.put(i)

    stop_thread(download_queue, download_threads)
    stop_thread(resize_queue, resize_threads)
    stop_thread(upload_queue, upload_threads)

    print("完成：", done_queue.qsize())


if __name__ == '__main__':
    main()
